package com.mccken.rapid.core.netty;

import java.net.InetSocketAddress;

import com.mccken.rapid.common.util.RemotingHelper;
import com.mccken.rapid.common.util.RemotingUtil;
import com.mccken.rapid.core.LifeCycle;
import com.mccken.rapid.core.RapidConfig;
import com.mccken.rapid.core.netty.processor.NettyProcessor;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.HttpServerExpectContinueHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;

/**
 * @program: rapid_gateway
 * @description:
 * @author: mccken
 * @create: 2022-01-03 18:52
 **/
@Slf4j
public class NettyHttpServer implements LifeCycle {
	
	private final RapidConfig rapidConfig;
	
	private final NettyProcessor nettyProcessor;
	
	private int port = 8888;
	
	private ServerBootstrap serverBootstrap;
	
	private EventLoopGroup eventLoopGroupBoss;
	
	private EventLoopGroup eventLoopGroupWork;
	
	public NettyHttpServer(RapidConfig rapidConfig, NettyProcessor nettyProcessor) {
		this.rapidConfig = rapidConfig;
		this.nettyProcessor = nettyProcessor;
		
		if (rapidConfig.getPort() > 0 && rapidConfig.getPort() < 65535) {
			this.port = rapidConfig.getPort();
		}
		
		init();
	}
	
	
	@Override
	public void init() {
		this.serverBootstrap = new ServerBootstrap();
		if ((useEPoll())) {
			this.eventLoopGroupBoss = new EpollEventLoopGroup(rapidConfig.getEventLoopGroupBossNum(),
					new DefaultThreadFactory("NettyBossEPoll"));
			this.eventLoopGroupWork = new EpollEventLoopGroup(rapidConfig.getEventLoopGroupWorkNum(), 
					new DefaultThreadFactory("NettyWorkEPoll"));
		} else {
			this.eventLoopGroupBoss = new NioEventLoopGroup(rapidConfig.getEventLoopGroupBossNum(),
					new DefaultThreadFactory("NettyBossNio"));
			this.eventLoopGroupWork = new NioEventLoopGroup(rapidConfig.getEventLoopGroupWorkNum(),
					new DefaultThreadFactory("NettyWorkNio"));
		}
	}
	
	@Override
	public void start() {
		ServerBootstrap handler = this.serverBootstrap
				.group(eventLoopGroupBoss, eventLoopGroupWork)
				.channel(useEPoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
				.option(ChannelOption.SO_BACKLOG, 1024)  // sync + accept = backlog
				.option(ChannelOption.SO_REUSEADDR, true)    // tcp端口重新绑定
				.option(ChannelOption.SO_KEEPALIVE, false)   // 如果在2h之内没有数据通信，TCP会自动发送一个活动探测数据报文
				.childOption(ChannelOption.TCP_NODELAY, true) // 该参数左右就是禁用Nagle算法，使用小数据传输时合并
				.childOption(ChannelOption.SO_SNDBUF, 65535) // 设置发送数据缓冲区大小
				.childOption(ChannelOption.SO_RCVBUF, 65535) // 设置接收数据缓冲区大小
				.localAddress(new InetSocketAddress(this.port))
				.childHandler(new ChannelInitializer<Channel>() {

					@Override
					protected void initChannel(Channel channel) {
						channel.pipeline().addLast(
								new HttpServerCodec(),
								new HttpObjectAggregator(rapidConfig.getMaxContentLength()),
								new HttpServerExpectContinueHandler(),
								new NettyServerConnectManagerHandler(),    // 连接管理器类
								new NettyHttpServerHandler(nettyProcessor)    // 核心逻辑，Netty HTTP 核心处理类
						);
					}
				});
		
		if (rapidConfig.isNettyAllocator()) {
			handler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
		}
		
		try {
			this.serverBootstrap.bind().sync();
			log.info("< ============= Rapid Server StartUp on Port: " + this.port + " ================ >");
		} catch (Exception e) {
			throw new RuntimeException("this.serverBootstrap.bind().sync() fail!", e);
		}

	}

	@Override
	public void shutdown() {
		if (eventLoopGroupWork != null) {
			eventLoopGroupWork.shutdownGracefully();
		}
		
		if (eventLoopGroupBoss != null) {
			eventLoopGroupBoss.shutdownGracefully();
		}
	}
	
	public EventLoopGroup getEventLoopGroupWork() {
		return eventLoopGroupWork;
	}
	
	public boolean useEPoll() {
		return rapidConfig.isUseEPoll() 
				&& RemotingUtil.isLinuxPlatform()
				&& Epoll.isAvailable();
	}

	/**
	 * 连接管理器
	 *  关于注册相关的，
	 */
	static class NettyServerConnectManagerHandler extends ChannelDuplexHandler {

		@Override
		public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
			final String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
			log.debug("NETTY SERVER PIPLINE: channelRegistered {}", remoteAddr);
			super.channelRegistered(ctx);
		}

		@Override
		public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
			final String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
			log.debug("NETTY SERVER PIPLINE: channelUnregistered {}", remoteAddr);
			super.channelUnregistered(ctx);
		}

		@Override
		public void channelActive(ChannelHandlerContext ctx) throws Exception {
			final String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
			log.debug("NETTY SERVER PIPLINE: channelActive {}", remoteAddr);
			super.channelActive(ctx);
		}

		@Override
		public void channelInactive(ChannelHandlerContext ctx) throws Exception {
			final String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
			log.debug("NETTY SERVER PIPLINE: channelInactive {}", remoteAddr);
			super.channelInactive(ctx);
		}

		/**
		 * 事件触发时
		 *  如果evt空闲时,关闭channel
		 *
		 */
		@Override
		public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
			if (evt instanceof IdleStateEvent) {
				IdleStateEvent event = (IdleStateEvent) evt;
				if (event.state().equals(IdleState.ALL_IDLE)){
					String remoteAddr = RemotingUtil.parseRemoteAddress(ctx.channel());
					log.warn("NETTY SERVER PIPLINE: userEventTriggered: IDLE {}", remoteAddr);
					ctx.channel().close();
				}
			}
			super.userEventTriggered(ctx, evt);
		}

		/**
		 * 抛异常时
		 *
		 */
		@Override
		public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
			final String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
			log.warn("NETTY SERVER PIPLINE: remoteAddr： {}, exceptionCaught {}", remoteAddr, cause);
			super.exceptionCaught(ctx, cause);
		}
	}
}
